package com.hzh.MLlib

import org.apache.spark.ml.linalg.Vectors
import org.apache.spark.sql.expressions.UserDefinedFunction
import org.apache.spark.sql.{DataFrame, SparkSession}

/**
 * 特征工程，多分类
 */
object Demo2Photos {
  def main(args: Array[String]): Unit = {
    /**
     * 创建环境
     *
     */

    val spark: SparkSession = SparkSession
      .builder()
      .config("spark.sql.shuffle.partitions", 1)
      .master("local[6]")
      .appName("Demo2Photos")
      .getOrCreate()

    import org.apache.spark.sql.functions._
    import spark.implicits._

    val imageData: DataFrame = spark
      .read
      .format("image")
      .load("D:\\bigdata培训\\spark\\day07\\resources\\train")

    //读取标记文件
    val nameAndLabelDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("name STRING,label Double")
      .load("data/train.txt")


    val comData: UserDefinedFunction = udf((data: Array[Byte]) => {
      val features: Array[Double] = data.map(byte => byte.toInt)
        .map(x => {
          if (x >= 0) {
            0.0
          } else {
            1.0
          }
        })
      //返回特征值
      Vectors.dense(features).toSparse
    })

    //自定义函数取出文件名称
    val comName: UserDefinedFunction = udf((path: String) => {
      val comName: String = path.split("/").last
      comName
    })


    val trainDF: DataFrame = imageData
      .select(comName($"image.origin") as "name", comData($"image.data") as "features") //取出文件名和数据
      //关联获取图片的一个标记 hint("broadcast")将小表广播出去
      .join(nameAndLabelDF.hint("broadcast"), "name")
      //取出目标值和特征向量
      .select($"label", $"features")


    //将数据保存为svm格式
    trainDF
      .write
      .format("libsvm")
      .save("data/image_data")


  }
}
